package com.chenxu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.chenxu.gmall.realtime.app.func.DimSink;
import com.chenxu.gmall.realtime.app.func.TableProcessFunction1;
import com.chenxu.gmall.realtime.bean.TableProcess;
import com.chenxu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Date:2021/07/11
 * Desc:准备用户业务的DWD层
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.准备环境
        //1.1 创建Flink流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1.2 设置并行度
        env.setParallelism(1);

        //1.3设置checkpoint
        //这里开启checkpoint的情况下，运行时必须开启HDFS；
        //System.setProperty("HADOOP_USER_NAME","chenxu");
        //每5000ms开启一次checkpoint
//        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
//        //超过1分钟失效
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        //设置状态后端，这里用文件系统；
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/checkpoint/basedbApp"));

        //TODO 从kafka的ODS层读取业务数据
        //业务数据主题；
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_group";

        //2.1 通过工具类获取Kafka的消费者
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> jsonStrDS = env.addSource(kafkaSource);

        //log测试；
        //jsonStrDS.print("json>>>>");
        //jsonStrDS.print("转换前");

        //TODO 3.对DS中数据进行结构的转换      String-->Json
        SingleOutputStreamOperator<JSONObject> jsonObjDS = jsonStrDS.map(jsonStr -> JSON.parseObject(jsonStr));
        //也可以用方法调用的方式,这种写法可以防止泛型擦除；
        //SingleOutputStreamOperator<JSONObject> jsonObjDS = jsonStrDS.map(JSON::parseObject);

        //jsonObjDS.print("转换后");

        /*
        数据大概样式：
        {"database":"gmall0709","table":"order_info","type":"update","ts":1626008658,"xid":254,"xoffset":4651,
        "data":{"id":28993,"consignee":"蒋莲真","consignee_tel":"13494023727","total_amount":21996.00,
        "order_status":"1001","user_id":438,"payment_way":null,"delivery_address":"第17大街第39号楼1单元977门",
        "order_comment":"描述178284","out_trade_no":"352538349474172","trade_body":"十月稻田 长粒香大米 东北大米 东北香米 5kg等10件商品",
        "create_time":"2021-07-11 21:04:17","operate_time":"2021-07-11 21:04:18","expire_time":"2021-07-11 21:19:17",
        "process_status":null,"tracking_no":null,"parent_order_id":null,"img_url":"http://img.gmall.com/646283.jpg",
        "province_id":29,"activity_reduce_amount":0.00,"coupon_reduce_amount":0.00,"original_total_amount":21989.00,
        "feight_fee":7.00,"feight_fee_reduce":null,"refundable_time":null},
        "old":{"operate_time":null}}
        有些数据不全
         */
        //TODO 4.对数据进行ETL   如果table为空 或者 data为空，或者长度<3  ，将这样的数据过滤掉
        //这里认为data中数据小于3的属于无用数据；
        SingleOutputStreamOperator<JSONObject> filteredDS = jsonObjDS.filter(
                jsonObj -> {
                    boolean flag = jsonObj.getString("table") != null
                            && jsonObj.getJSONObject("data") != null
                            && jsonObj.getJSONObject("data").size() >= 3;
                    return flag;
                });

        //测试：        开启zk.sh、kk.sh、maxwell.sh、当前应用程序；再去开启业务数据jar包：
        //        /opt/module/rt_dblog $java -jar gmall2020-mock-db-2020-11-27.jar
        //filteredDS.print("------");

        //TODO 5. 动态分流  事实表放到主流，写回到kafka的DWD层；如果维度表，通过侧输出流，写入到Hbase
        //5.1定义输出到Hbase的侧输出流标签
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>(TableProcess.SINK_TYPE_HBASE){};

        //5.2 主流 写回到Kafka的数据
        //process是最底层的处理函数，侧输出流的输出用这个函数是最方便的；

        //为什么不用下面这种方法呢，因为这样的话就需要每一条过来的数据都去遍历一次MySql的配置信息表，这对于实时计算而言是不合理的；
//        filteredDS.process(
//                new ProcessFunction<JSONObject, JSONObject>() {
//                    OutputTag<JSONObject> Tag = new OutputTag<JSONObject>("xxxx");

        //如果重写Open函数来做，只查一次；
        //此时配置信息表并不是一成不变的，所以不好；
        //这里还有一种方法，可以设置定时器，比如每5秒来查一次；
//                    @Override
//                    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
//                        context.output(Tag,jsonObject);
//                    }
//                }
//        )

        //主流kafkaDS
        SingleOutputStreamOperator<JSONObject> kafkaDS = filteredDS.process(
                new TableProcessFunction1(hbaseTag)
        );

        //5.3获取侧输出流    写到Hbase的数据
        DataStream<JSONObject> hbaseDS = kafkaDS.getSideOutput(hbaseTag);

        //动态分流测试：
        kafkaDS.print("事实>>>>");
        hbaseDS.print("维度>>>>");

        //测试；
        //开启zk、kk、hdfs、hbase、phoenix、maxwell、本应用程序，记得提前在gmall0709_realtime数据库中手动插入一条配置文件信息；
        //此时phoenix会自动建表GMALL0709_REALTIME
        //手动修改gmall0709数据库中base_trademark表的数据，应该会给出类似下面的输出；
        //维度>>>>:2> {"database":"gmall0709","xid":18593,"data":{"id":17},"commit":true,"sink_table":"DIM_BASE_TRADEMARK","type":"insert","table":"base_trademark","ts":1626230909}
        //hbase中不会接收到具体数据；

        //再把刚才配置表中的sink_type改为kafka，把sink_table改为first
        //此时Map会重新读取这条新的配置信息，加入到内存中，数据也会输出到kafka中；
        //重启应用程序，会出现类似下面的输出：
        //事实>>>>:4> {"database":"gmall0709","xid":19366,"data":{"id":19},"commit":true,"sink_table":"first","type":"insert","table":"base_trademark","ts":1626231206}
        //但是kafka中不会接收到数据；

        //所以接下来要做的，就是把对应的维度数据再保存到hbase表中，把事实数据输出到kafka中；

        //TODO 6.将维度数据保存到Phoenix对应的维度表中
        //DimSink()这里用富函数来做，因为富函数的方法更多，有open方法可以来定义连接；
        hbaseDS.addSink(new DimSink());

        //测试
        //启动zk、kk、hdfs、hbase、maxwell、phoenix、应用程序；
        //手动修改刚才的配置表信息，改回hbase和DIM_BASE_TRADEMARK；
        //再去base_trademark表中添加一下数据；
        //输出一条信息如下：
        //维度>>>>:2> {"database":"gmall0709","xid":30932,"data":{"tm_name":"20","id":20},"commit":true,"sink_table":"DIM_BASE_TRADEMARK","type":"insert","table":"base_trademark","ts":1626240730}
        //向Phoenix插入数据的SQL:upsert into GMALL0709_REALTIME.DIM_BASE_TRADEMARK(tm_name,id) values ('20','20')
        //在phoenix中查询数据是否插入：
        //select * from GMALL0709_REALTIME.DIM_BASE_TRADEMARK;

        //TODO 7.将事实数据写回到kafka的dwd层
        //分析：将数据写入到kafka中，相当于这里缺一个生产者；所以调工具类的getKafkaSink方法，但这里这么写很难用；
        //1.这里的topic只能写一个,但是事实表中的数据不止会写入一个主题中；
        //2.kafkaDS数据中的数据类型不是字符串，而是JSONObject；
        //kafkaDS.addSink(MyKafkaUtil.getKafkaSink("XXX"));
        //所以这里选用的是可以支持多个topic的FlinkKafkaProducer对象；

        //可以发往多个主题；
        FlinkKafkaProducer<JSONObject> kafkaSink = MyKafkaUtil.getKafkaSinkBySchema(
                new KafkaSerializationSchema<JSONObject>() {
                    @Override
                    public void open(SerializationSchema.InitializationContext context) throws Exception {
                        System.out.println("kafka序列化");
                    }

                    //核心方法，每一条数据的序列化；
                    @Override
                    //这里的两个byte[], byte[]分别指代K,V；只不过这里没用K，只用了V；注意：K不是这里的topic，topic不传输；
                    //K（即Topic的作用是为了帮助分配到不同的主题中）
                    public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                        String sinkTopic = jsonObj.getString("sink_table");
                        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
                        //两个参数分别表示：主题名和数据值；
                        //序列化的意义就是把数据转化为字节数组，所以这里的value转化为Byte类型来做；主题找到位置即可，不用序列化；
                        return new ProducerRecord<>(sinkTopic,dataJsonObj.toString().getBytes());
                    }
                }
        );

        //数据测试：

        kafkaDS.addSink(kafkaSink);

        env.execute();
    }
}
